package com.huisou.dao;

import static org.apache.solr.handler.dataimport.DataImportHandlerException.SEVERE;
import static org.apache.solr.handler.dataimport.DataImportHandlerException.wrapAndThrow;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.solr.handler.dataimport.Context;
import org.apache.solr.handler.dataimport.DataImporter;
import org.apache.solr.handler.dataimport.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.huisou.utils.SpringContextHolder;

public class BoneCpDataSource extends DataSource<Iterator<Map<String, Object>>> {

	private static final Logger LOG = LoggerFactory.getLogger(BoneCpDataSource.class);
	
	javax.sql.DataSource ds;
	
	private long connLastUsed = 0;
	
	private Connection conn;

	@Override
	public void init(Context context, Properties initProps) {
		 LOG.warn("初始化数据源 " );
		 String coreName = context.getSolrCore().getName();
		 if("ask".equals(coreName))
			 ds = (javax.sql.DataSource)SpringContextHolder.getBean("askDataSource");
		 else
			 ds = (javax.sql.DataSource)SpringContextHolder.getBean("dataSource");
		 Object o = initProps.get(CONVERT_TYPE);
		    if (o != null)
		      convertType = Boolean.parseBoolean(o.toString());
		    String bsz = initProps.getProperty("batchSize");
		    if (bsz != null) {
		      bsz = context.replaceTokens(bsz);
		      try {
		        batchSize = Integer.parseInt(bsz);
		        if (batchSize == -1)
		          batchSize = Integer.MIN_VALUE;
		      } catch (NumberFormatException e) {
		        LOG.warn("Invalid batch size: " + bsz);
		      }
		    }

	    for (Map<String, String> map : context.getAllEntityFields()) {
	      String n = map.get(DataImporter.COLUMN);
	      String t = map.get(DataImporter.TYPE);
	      if ("sint".equals(t) || "integer".equals(t))
	        fieldNameVsType.put(n, Types.INTEGER);
	      else if ("slong".equals(t) || "long".equals(t))
	        fieldNameVsType.put(n, Types.BIGINT);
	      else if ("float".equals(t) || "sfloat".equals(t))
	        fieldNameVsType.put(n, Types.FLOAT);
	      else if ("double".equals(t) || "sdouble".equals(t))
	        fieldNameVsType.put(n, Types.DOUBLE);
	      else if ("date".equals(t))
	        fieldNameVsType.put(n, Types.DATE);
	      else if ("boolean".equals(t))
	        fieldNameVsType.put(n, Types.BOOLEAN);
	      else if ("binary".equals(t))
	        fieldNameVsType.put(n, Types.BLOB);
	      else
	        fieldNameVsType.put(n, Types.VARCHAR);
	    }
	}

	@Override
	public Iterator<Map<String, Object>> getData(String query) {
		ResultSetIterator r = new ResultSetIterator(query);
		return r.getIterator();
	}
	
	  private Connection getConnection() throws Exception {
		    LOG.debug("获取数据连接" );
		    long currTime = System.currentTimeMillis();
		    if (currTime - connLastUsed > CONN_TIME_OUT) {
		      synchronized (this) {
		    	LOG.debug("获取新的数据连接" );
		    	Connection tmpConn = ds.getConnection();
		        closeConnection();
		        connLastUsed = System.currentTimeMillis();
		        return conn = tmpConn;
		      }
	
		    } else {
		      LOG.debug("使用已有的数据连接" );
		      connLastUsed = currTime;
		      return conn;
		    }
		  }

	  @Override
	  public void close() {
	    try {
	      closeConnection();
	    } finally {
	      isClosed = true;
	    }
	  }

	  private void closeConnection()  {
	    try {
	    	LOG.debug("关闭数据连接" );
	      if (conn != null) {
	        conn.close();
	      }
	    } catch (Exception e) {
	      LOG.error("Ignoring Error when closing connection", e);
	    }
	  }

	private Map<String, Integer> fieldNameVsType = new HashMap<String, Integer>();
	
	private boolean isClosed = false;

	private boolean convertType = false;

	private int batchSize = FETCH_SIZE;

	private int maxRows = 0;

	private class ResultSetIterator {
		ResultSet resultSet;

		Statement stmt = null;

		List<String> colNames;

		Iterator<Map<String, Object>> rSetIterator;

		public ResultSetIterator(String query) {

			try {
				Connection c = getConnection();
				stmt = c.createStatement(ResultSet.TYPE_FORWARD_ONLY,
						ResultSet.CONCUR_READ_ONLY);
				stmt.setFetchSize(batchSize);
				stmt.setMaxRows(maxRows);
				LOG.debug("Executing SQL: " + query);
				long start = System.currentTimeMillis();
				if (stmt.execute(query)) {
					resultSet = stmt.getResultSet();
				}
				LOG.trace("Time taken for sql :"
						+ (System.currentTimeMillis() - start));
				colNames = readFieldNames(resultSet.getMetaData());
			} catch (Exception e) {
				wrapAndThrow(SEVERE, e, "Unable to execute query: " + query);
			}
			if (resultSet == null) {
				rSetIterator = new ArrayList<Map<String, Object>>().iterator();
				return;
			}

			rSetIterator = new Iterator<Map<String, Object>>() {
				public boolean hasNext() {
					return hasnext();
				}

				public Map<String, Object> next() {
					return getARow();
				}

				public void remove() {/* do nothing */
				}
			};
		}

		private Iterator<Map<String, Object>> getIterator() {
			return rSetIterator;
		}

		private Map<String, Object> getARow() {
			if (resultSet == null)
				return null;
			Map<String, Object> result = new HashMap<String, Object>();
			for (String colName : colNames) {
				try {
					if (!convertType) {
						// Use underlying database's type information
						result.put(colName, resultSet.getObject(colName));
						continue;
					}

					Integer type = fieldNameVsType.get(colName);
					if (type == null)
						type = Types.VARCHAR;
					switch (type) {
					case Types.INTEGER:
						result.put(colName, resultSet.getInt(colName));
						break;
					case Types.FLOAT:
						result.put(colName, resultSet.getFloat(colName));
						break;
					case Types.BIGINT:
						result.put(colName, resultSet.getLong(colName));
						break;
					case Types.DOUBLE:
						result.put(colName, resultSet.getDouble(colName));
						break;
					case Types.DATE:
						result.put(colName, resultSet.getDate(colName));
						break;
					case Types.BOOLEAN:
						result.put(colName, resultSet.getBoolean(colName));
						break;
					case Types.BLOB:
						result.put(colName, resultSet.getBytes(colName));
						break;
					default:
						result.put(colName, resultSet.getString(colName));
						break;
					}
				} catch (SQLException e) {
					logError("Error reading data ", e);
					wrapAndThrow(SEVERE, e, "Error reading data from database");
				}
			}
			return result;
		}

		private boolean hasnext() {
			if (resultSet == null)
				return false;
			try {
				if (resultSet.next()) {
					return true;
				} else {
					close();
					return false;
				}
			} catch (SQLException e) {
				close();
				wrapAndThrow(SEVERE, e);
				return false;
			}
		}

		private void close() {
			try {
				if (resultSet != null)
					resultSet.close();
				if (stmt != null)
					stmt.close();
			} catch (Exception e) {
				logError("Exception while closing result set", e);
			} finally {
				resultSet = null;
				stmt = null;
			}
		}
	}

	private List<String> readFieldNames(ResultSetMetaData metaData)
			throws SQLException {
		List<String> colNames = new ArrayList<String>();
		int count = metaData.getColumnCount();
		for (int i = 0; i < count; i++) {
			colNames.add(metaData.getColumnLabel(i + 1));
		}
		return colNames;
	}
	
	@Override
	  protected void finalize() throws Throwable {
	    try {
	      if(!isClosed){
	        LOG.error("在对象销毁之前进行最后一次关闭连接的尝试");
	        close();
	      }
	    } finally {
	      super.finalize();
	    }
	  }

	private static final long CONN_TIME_OUT = 72 * 1000; // 10 seconds

	private static final int FETCH_SIZE = 500;

	public static final String URL = "url";

	public static final String JNDI_NAME = "jndiName";

	public static final String DRIVER = "driver";

	public static final String CONVERT_TYPE = "convertType";

	private void logError(String msg, Exception e) {
		LOG.warn(msg, e);
	}
}